package com.rent.recommend

import com.mongodb.casbah.{MongoClient, MongoClientURI}
import com.mongodb.casbah.commons.MongoDBObject
import com.rent.recommend.AddHouse.MONGODB_HOUSES_COLLECTION
import org.apache.log4j.Logger
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}

/*
@李子宁
Happy,happy everyday!
冲鸭！
2021/9/30

*/
//private case class MongoConfig(uri:String, db:String)
//
//private case class ESConfig(httpHosts:String, transportHosts:String, index:String, clustername:String)
//
//private case class House(hid: Int, houseSingleUrl: String, title: String, singleType: String, size: String, directType: String, singleLayer: String, rentType: String, place: String, underPlace: String, schoolPlace: String , price: Int)
//private case class Hplace(pid: Int , name: String, lon: Double , lai: Double)
object TestPrin {
  val HOUSE_PLACE_PATH = "D:\\IntelliJ IDEA-workspace\\HouseRent\\HouseRecommend\\DataLoader\\src\\main\\resources\\fils\\ftx.csv";
  val ES_HOUSE_INDEX = "house"
  val MONGODB_HOUSE_COLLECTION = "House"
  val log = Logger.getLogger(TestPrin.getClass)
  def main(args: Array[String]): Unit = {

    val config = Map(
      "spark.cores" -> "local[*]",
      "es.httpHosts" -> "localhost:9200",
      "es.transportHosts" -> "localhost:9300",
      "es.index" -> "house",
      "es.cluster.name" -> "elasticsearch",
      "mongo.uri" -> "mongodb://localhost:27017/houserec",
      "mongo.db" -> "houserec"
    )

    // 创建一个sparkConf
    val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("DataLoader")

    // 创建一个SparkSession
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    import spark.implicits._
    // 声明了一个 ES 配置的隐式参数
    // implicit val esConfig = ESConfig(config("es.httpHosts"), config("es.transportHosts"), config("es.index"), config("es.cluster.name"))
    //    val houseDF = spark.sql("select hid,")
    var index = 1922
    implicit val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))
    val houseRDD = spark.sparkContext.textFile(HOUSE_PLACE_PATH)
    val houseDF = houseRDD.map(item => {
      val attr = item.split(",")
      val hid = index
      index += 1
      val types = attr(2).replace("\n","").replace("\"","").replace(" ","").trim.split("\\|")
        log.info(attr)
      //hid: Int, title: String, singleType: String, size: String, directType: String, singleLayer: String, rentType: String, place: String, underPlace: String, schoolPlace: String , price: Int)

      //明发滨江新城一期 3室1卫1厅 112精装修 2300元/月	https://nanjing.zu.fang.com/chuzu/3_183261245_1.htm	"
      //                整租|3室1厅|112㎡|朝南
      //            "	浦口-桥北-明发滨江新城一期	1960		https://cdnsfb.soufunimg.com/viewimage/sfb/houseverfiy/2021_9/19/M9/46/4c5a4e29-12a7-4866-990b-09ee6163907c/275x207.jpg
     /* House(hid.toInt,
        //房子的连接
        attr(1).trim,
        //房子名称
        attr(0).trim,
        //房子的房型，几室几厅
        types(1).trim,
        //房子的面积
        types(2).trim,
        //房子的装修类型
        "",
        //房子的楼层
        "",
        //房子的出租类型
        types(0).trim,
        //房子的地理位置
        attr(3).trim,
        //房子是否靠近地铁
        attr(5).trim,
        //房子是否靠近学校
        "",
        //房子的价格
        attr(4).toInt)*/
    })//.toDF()

    // 将数据保存到MongoDB
    //   storeDataInMongoDB(houseDF)

     // storeDataInES(houseDF)

    // 关闭 Spark
    spark.stop()

  }

  def storeDataInES(movieDF:DataFrame): Unit = {
    movieDF
      .write
      .option("es.nodes","localhost:9200")
      .option("es.http.timeout","100m")
      .option("es.mapping.id","hid")
      .mode("append")
      .format("org.elasticsearch.spark.sql")
      .save("house"+"/"+ES_HOUSE_INDEX)
  }

  def storeDataInMongoDB(browseDF: DataFrame)(implicit mongoConfig: MongoConfig): Unit = {
    // 新建一个mongodb的连接
    val mongoClient = MongoClient(MongoClientURI(mongoConfig.uri))

    // 如果mongodb中已经有相应的数据库，先删除
    mongoClient(mongoConfig.db)(MONGODB_HOUSE_COLLECTION).dropCollection()
    // 将DF数据写入对应的mongodb表中
    browseDF.write
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_HOUSE_COLLECTION)
      .mode("append")
      .format("com.mongodb.spark.sql")
      .save()

    //对数据表建索引
    mongoClient(mongoConfig.db)(MONGODB_HOUSE_COLLECTION).createIndex(MongoDBObject("hid" -> 1))


    mongoClient.close()

  }
}
